package en.edu.zxj.mq.mqserver;

import en.edu.zxj.mq.common.Consumer;
import en.edu.zxj.mq.common.MqException;
import en.edu.zxj.mq.common.Router;
import en.edu.zxj.mq.mqserver.core.*;
import en.edu.zxj.mq.mqserver.datacenter.DiskDataCenter;
import en.edu.zxj.mq.mqserver.datacenter.MemoryDataCenter;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description：通过这个类, 来标识虚拟主机
 * 每个虚拟主机下面都管理着自己的 交换机, 队列, 绑定, 消息, 数据
 * 同时提供 API 供上层调用
 * 针对 VirtualHost 这个类, 作为业务逻辑的整合者, 就需要对于代码中抛出的异常进行处理了
 *
 * @author: zxj
 * @date: 2024-03-01
 * @time: 19:23:34
 */
@Getter
@Slf4j
public class VirtualHost {
    private final String virtualHostName;
    private final DiskDataCenter diskDataCenter = new DiskDataCenter();
    private final MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    // Router 定义转发规则
    private final Router router = new Router();
    // ConsumerManager 实现消费逻辑
    private final ConsumerManager consumerManager = new ConsumerManager(this);

    // 锁
    private final Object lockerQueue = new Object();
    private final Object lockerExchange = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        // 先初始化硬盘操作
        diskDataCenter.init();
        // 后初始化内存操作
        memoryDataCenter.init();

        // 从磁盘中恢复数据到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (Exception e) {
            log.error("从磁盘中恢复数据到内存失败!");
            e.printStackTrace();
            return;
        }

        log.info("初始化 VirtualHost 成功, VirtualHostName: {}", virtualHostName);
    }


    /**
     * @description: 创建交换机, declare 表示存在就不创建, 因此不叫做 "exchangeCreate"
     * 此处的 autoDelete, arguments 其实并没有使用, 只是先预留出来. (RabbitMQ 是支持的)
     * 约定, 交换机/队列的名字, 都加上 VirtualHostName 作为前缀, 这样不同 VirtualHost 中就可以存在同名的交换机或者队列了
     * 先写磁盘, 后写内存, 因为写磁盘失败概率更大, 如果磁盘写失败了, 也就不必要写内存了
     * @param: [exchangeName, exchangeType, durable, autoDelete, arguments]
     * @return: 抛异常就返回 false, 正常执行逻辑就返回 true
     **/
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        // 真实的 exchangeName 要加上 virtualHostName 前缀
        exchangeName = virtualHostName + exchangeName;
        synchronized (lockerExchange) {

            try {
                // 1. 判断交换机是否存在
                Exchange exsitsExchange = memoryDataCenter.getExchange(exchangeName);
                if (exsitsExchange != null) {
                    log.info("交换机已经存在, exchangeName: {}, exchangeType: {}", exchangeName, exchangeType);
                    return true;
                }

                // 2. 构造 Exchange 对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                // 3. 先写入磁盘
                if (exchange.getDurable()) {
                    diskDataCenter.insertExchange(exchange);
                }

                // 4. 后写入内存
                memoryDataCenter.insertExchange(exchange);

                log.info("交换机创建成功, exchangeName: {}, exchangeType: {}", exchangeName, exchangeType);

                return true;
            } catch (Exception e) {
                log.warn("创建交换机失败, exchangeName: {}, exchangeType: {}", exchangeName, exchangeType);
                e.printStackTrace();
                return false;
            }
        }
    }


    /**
     * @description: 删除交换机
     **/
    public boolean exchangeDelete(String exchangeName) {
        // 真正存储的 exchangeName
        exchangeName = virtualHostName + exchangeName;
        synchronized (lockerExchange) {
            try {
                // 1. 判断交换机是否存在
                Exchange exsitsExchange = memoryDataCenter.getExchange(exchangeName);
                if (exsitsExchange == null) {
                    throw new MqException("交换机不存在, 无法删除! exchangeName; " + exchangeName);
                }
                // 2. 删除磁盘中的交换机
                diskDataCenter.deleteExchangeByName(exchangeName);
                // 3. 删除内存中的交换机
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("删除交换机成功, exchangeName: {},", exchangeName);
                return true;
            } catch (Exception e) {
                log.warn("删除交换机失败, exchangeName: {},", exchangeName);
                e.printStackTrace();
                return false;
            }
        }
    }


    /**
     * @description: 创建队列
     **/
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        // 真实的 queueName
        queueName = virtualHostName + queueName;
        synchronized (lockerQueue) {

            try {
                // 1. 判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getMSGQueue(queueName);
                if (existsQueue != null) {
                    log.info("队列已经存在, queueName: {}", queueName);
                    return true;
                }
                // 2. 创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 先存入磁盘
                if (queue.getDurable()) {
                    diskDataCenter.insertMSGQueue(queue);
                }
                // 4. 后存入到内存
                memoryDataCenter.insertMSGQueue(queue);
                log.info("队列创建成功, queueName: {}", queueName);
                return true;
            } catch (Exception e) {
                log.warn("创建队列失败, queueName: {}", queueName);
                e.printStackTrace();
                return false;
            }
        }
    }

    /**
     * @description: 删除队列
     **/
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        synchronized (lockerQueue) {
            try {
                // 1. 判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getMSGQueue(queueName);
                if (existsQueue == null) {
                    throw new MqException("要删除的队列不存在, 无法删除! queueName=" + queueName);
                }
                // 2. 删除磁盘中的队列
                if (existsQueue.getDurable()) {
                    diskDataCenter.deleteMSGQueueByName(queueName);
                }
                // 3. 删除内存中的队列
                memoryDataCenter.deleteMSGQueue(queueName);
                log.info("删除队列成功, queueName: {}", queueName);

                return true;
            } catch (Exception e) {
                log.warn("删除队列失败, queueName: {}", queueName);
                e.printStackTrace();
                return false;
            }
        }
    }

    /**
     * @description: 添加绑定
     **/
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        // 加上 virtualHostName 前缀
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        synchronized (lockerQueue) {
            synchronized (lockerExchange) {
                try {
                    // 1. 判断绑定是否存在
                    Binding existedBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existedBinding != null) {
                        log.info("绑定存在, queueName: {}, exchangeName: {}, bindingKey: {}", queueName, exchangeName, bindingKey);
                        return true;
                    }
                    // 2. 判断 bindingKey 是否合法
                    if (!router.checkBindingKeyValid(bindingKey)) {
                        throw new MqException("bindingKey 不合法! bindingKey: " + bindingKey);
                    }
                    // 3. 创建绑定
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取绑定对应的队列和交换机, 判断这两个是否存在, 都存在才能创建
                    MSGQueue msgQueue = memoryDataCenter.getMSGQueue(queueName);
                    if (msgQueue == null) {
                        throw new MqException("队列不存在, queueName: " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("交换机不存在, exchangeName: " + exchangeName);
                    }
                    // 5. 写入磁盘
                    if (msgQueue.getDurable() && exchange.getDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 写入内存
                    memoryDataCenter.insetBinding(binding);
                    log.info("添加绑定成功, queueName: {}, exchangeName: {}, bindingKey: {}", queueName, exchangeName, bindingKey);
                    return true;
                } catch (Exception e) {
                    log.warn("添加绑定失败, queueName: {}, exchangeName: {}, bindingKey: {}", queueName, exchangeName, bindingKey);
                    e.printStackTrace();
                    return false;
                }
            }
        }
    }

    /**
     * @description: 删除绑定
     **/
    public boolean queueUnBind(String queueName, String exchangeName) {
        // 加上 virtualHostName 前缀
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        synchronized (lockerExchange) {
            synchronized (lockerQueue) {

                try {
                    // 1. 判断绑定是否存在
                    Binding existedBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existedBinding == null) {
                        throw new MqException("要删除的绑定不存在, 无法删除, exchangeName: " + exchangeName + " , queueName: " + queueName);
                    }
                    // 2. 无论绑定是否持久化了, 都试着删除一下磁盘中的数据, 影响不大
                    diskDataCenter.deleteBinding(exchangeName, queueName);
                    // 3. 删除内存
                    memoryDataCenter.deleteBinding(existedBinding);
                    log.info("删除绑定成功, queueName: {}, exchangeName: {}", queueName, exchangeName);
                    return true;
                } catch (Exception e) {
                    log.warn("删除绑定失败, queueName: {}, exchangeName: {}", queueName, exchangeName);
                    e.printStackTrace();
                    return false;
                }
            }
        }
    }

    /**
     * @description: 发布消息
     * 发布消息其实就是把消息发送给指定的exchange, 再根据 Exchange 和 Queue 的 Binding 关系, 转发到对应队列中
     * 发送消息需要指定 routingKey, 这个值的作用和 ExchangeType 相关的
     * Direct: routingKey 就是对应的队列名字, 此时不需要 binding 关系, 也不需要 bindingKey, 就可以直接转发消息
     * Fanout: routingKey 不起作用, bindingKey 也不起作用, 此时消息会转发给绑定该交换机上的所有队列中
     * Topic: routingKey 是一个特定的字符串, 会和 bindingKey 按照一定规则进行匹配, 如果匹配成功, 则发送到对应的队列中, 具体规则在 Router 类中介绍
     * @param: [exchangeName, routingKey, basicProperties 消息的元消息, body 消息本体]
     * @return:
     **/
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            // 1. 转换交换机的名字, 如果为 null, 就使用默认的交换机名字
            if (exchangeName == null) {
                exchangeName = "";
            }
            exchangeName = virtualHostName + exchangeName;

            // 2. 检验 routingKey 的合法性
            if (!router.checkRoutingKeyValid(routingKey)) {
                throw new MqException("routingKey 非法! routingKey: " + routingKey);
            }

            // 3. 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("交换机不存在! exchangeName: " + exchangeName);
            }

            // 4. 依据交换机的类型来进行消息转发
            if (exchange.getType() == ExchangeType.DIRECT) {
                // 按照直接交换机的方式来转发消息
                // 此时 routingKey 作为队列的名字, 直接把消息写入指定的队里中, 此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                // 5. 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 6. 找找队列名字对应的对象
                MSGQueue msgQueue = memoryDataCenter.getMSGQueue(queueName);
                if (msgQueue == null) {
                    throw new MqException("队列不存在, queueName=" + queueName);
                }
                // 7. 队列存在, 直接给队列中写入消息 -- 执行一次方法就消费一次消息
                sendMessage(msgQueue, message);
            } else {
                // 按照 fanout 和 topic 的方式来转发
                // 5. 找到该交换机关联的所有绑定, 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // ① 获取到绑定对象, 判断对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue msgQueue = memoryDataCenter.getMSGQueue(binding.getQueueName());
                    if (msgQueue == null) {
                        // 此处就不抛异常, 可能此处有多个这样的队列
                        // 希望不要因为一个队列的失败, 影响到其他队列的消息的传输
                        log.warn("basicPublish 发送消息是, 发现队列不存在! queueName: {}", binding.getQueueName());
                        continue;
                    }
                    // ② 构造消息对象, 发送给每一个队列的对象都是一个新的复制体
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // ③ 判定这个消息是否能转发给改队列
                    //    如果是 fanout, 所有绑定的队列都是要转发的
                    //    如果是 topic, 还需要判定下, bindingKey 和 routingKey 是不是匹配
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    // ④ 真正的转发消息给队列
                    sendMessage(msgQueue, message);
                }
            }
            log.info("发送信息成功, exchangeName: {}, routingKey: {}", exchangeName, routingKey);
            return true;
        } catch (Exception e) {
            log.warn("发送信息失败, exchangeName: {}, routingKey: {}", exchangeName, routingKey);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * @description: 发送一次消息
     **/
    private void sendMessage(MSGQueue msgQueue, Message message) throws IOException, MqException, InterruptedException {
        // 此处发送消息, 就是把消息写入到 硬盘 和 内存 上
        int deliverMode = message.getDeliverMode();
        // deliverMode 为 1, 不持久化, 为 2 表示持久化
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(msgQueue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(msgQueue, message);

        // 通知消费者可以消费消息, 就是让消费者从对应的内存中取出消息
        consumerManager.notifyConsume(msgQueue.getName());
    }


    /**
     * @description: 订阅消息
     * 添加一个队列的订阅者, 当队列收到消息之后, 就要把消息推送给对应的订阅者
     * @param: [consumerTag: 消费者的身份标识,
     * queueName: 订阅的队列名字,
     * autoAck: 消息被消费后的应当方式, true 为自动应当, false 为手动应答
     * consumer: 是一个回调函数, 此处类型设定成函数是接口, 这样后续调用 basicConsume 并且传递实参的时候, 就可以写成 lambda 样子了]
     * @return: true 表示订阅成功, false 表示订阅失败
     **/
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象, 把这个对应的队列找到, 再把这个 Consumer 对象添加到该队列中.
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            log.info("basicConsume 成功! consumerTag: {}, queueName: {}", consumerTag, queueName);
            return true;
        } catch (Exception e) {
            log.warn("basicConsume 失败! consumerTag: {}, queueName: {}", consumerTag, queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * @description: 消息确认
     **/
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            // 1. 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("要确认的消息不存在, messageId: " + messageId);
            }
            MSGQueue msgQueue = memoryDataCenter.getMSGQueue(queueName);
            if (msgQueue == null) {
                throw new MqException("要确认的队列不存在, queueName: " + queueName);
            }
            // 2. 删除硬盘上的数据
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(msgQueue,message);
            }
            // 3. 删除消息中心中心的数据
            memoryDataCenter.deleteMessage(messageId);
            // 4. 删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            log.info("basicAck 成功, 消息确认成功! queueName: {}, messageId: {}",queueName,messageId);
            return true;
        } catch (Exception e) {
            log.warn("basicAck 失败, 消息确认失败! queueName: {}, messageId: {}",queueName,messageId);
            e.printStackTrace();
            return false;
        }
    }


}
